package dmp.topk;

import java.io.IOException;
import java.util.TreeMap;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopkReducer extends Reducer<Text, Text, Text, Text>{
	private TreeMap<Long, String> map=new TreeMap<>();
	private Text  resultvalue=new Text();
	
	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		String url;
		Long count;
		for (Text value : values) {
			url=value.toString().split("\t")[0];
			count=Long.parseLong(value.toString().split("\t")[1]);
			map.put(count, url);
			if(map.size()>2){
				map.remove(map.firstKey());
			}
		}
		StringBuilder sb=new StringBuilder();
		for(String each: map.values()){
			sb.append(each).append("|");
		}
		resultvalue.set(new Text(sb.substring(0, sb.length()-1)));
		context.write(key,resultvalue);
	}
	
}
